package com.etlmaster.executor.steps.file;


import com.etlmaster.executor.Step;
import com.etlmaster.executor.StepMeta;
import com.etlmaster.executor.annotation.StepInstruction;
import com.etlmaster.executor.processors.File.FileTransInfo;
import com.etlmaster.executor.processors.File.FileTransfer;
import com.etlmaster.executor.utils.StepState;
import com.etlmaster.executor.utils.StringUtils;
/**
 * 
 * @author MeiKefu
 * @date 2015-2-16
 */
@StepInstruction("fileTrans")
public class FileTransferStep extends Step {

	private FileTransferStepMeta meta;

	public FileTransferStep(StepMeta meta) {
		super(meta);
		this.meta = (FileTransferStepMeta) meta;
	}
	@Override
	public boolean stop() {
		return false;
	}

	@Override
	protected int handle() {
		FileTransInfo fileTransInfo = new FileTransInfo(
				meta.getSourceVfsPath(), meta.getTargetVfsPath());
		fileTransInfo.setAppend(meta.isAppend());
		fileTransInfo.setNeedChangeCharset(meta.needChangeChatset());
		fileTransInfo.setSourceCharSet(meta.getSourceCharset());
		fileTransInfo.setTargetCharSet(meta.getTargetCharset());
		fileTransInfo.setRowProcessScript(meta.getRowProcessScript());
		fileTransInfo.setRowRead(meta.isRowRead());
		
		fileTransInfo.setStartRowNum(meta.getStartRowNum());
		fileTransInfo.setKeepBack(meta.isKeepBack());
		fileTransInfo.setDeCompress(meta.isDeCompress());

		if(StringUtils.isNumberic(meta.getRowNum()))
			fileTransInfo.setSplitRowNum(Integer.parseInt(meta.getRowNum()));
		if(StringUtils.isNumberic(meta.getFileSize()))
			fileTransInfo.setSplitSize(Long.parseLong(meta.getFileSize()));
		if(StringUtils.isNumberic(meta.getFileMergeSize()))
			fileTransInfo.setFileMergeSize(Long.parseLong(meta.getFileMergeSize())*1024*1024);
		fileTransInfo.setSplitType(meta.getSplitType());
		FileTransfer fileTranser = new FileTransfer(fileTransInfo);

		boolean isSuccess = fileTranser.process();
		meta.getStepState().getReturnValue().put("fileTransInfo", fileTransInfo);
		meta.getStepState().getReturnValue().put("isMerged", fileTransInfo.isMerged());
		if (isSuccess) {
			return StepState.RUNNING_SUCCESSED;
		} else {
			return StepState.RUNNING_FAILED;
		}
	}
}
